package pf

import (
	"fmt"
	"os"
	"time"

	"go.manyji.com/lib/app"
	"go.manyji.com/lib/config"
	"go.manyji.com/lib/log"
	"go.manyji.com/lib/redis"
	"go.manyji.com/lib/worker"
	"go.manyji.com/sieve/msg"
)

var (
	rc = redis.Connect(&redis.Server{
		Addr:   *config.String("mq.addr", "127.0.0.1:6400", ""),
		Pwd:    *config.String("mq.pwd", "", ""),
		Prefix: *config.String("mq.prefix", "SIEVE/MQ/", ""),
	})

	msgEncode = *config.String("msg.encode", "", "")
	msgDecode = *config.String("msg.decode", "", "")
	fNum      = config.Int("worker.num", 10, "")
	w         = worker.NewWorkers(handleMsg)

	pfname  string
	handler func(string) (bool, error)
)

func init() {
	app.OnStart(func() {
		w.SetNum(*fNum)
		config.OnReload(func() { w.SetNum(*fNum) })
	})
}

func Run(pf string, fn func(string) (bool, error)) {
	if msgEncode != "" {
		fmt.Println((&msg.Msg{
			Pf:      pf,
			Account: msgEncode,
			TaskId:  int(time.Now().Unix()),
		}).Encode())
		os.Exit(0)
	}
	if msgDecode != "" {
		m := &msg.Msg{}
		err := m.Decode(msgDecode)
		fmt.Println(m, "  ", err)
		os.Exit(0)
	}
	pfname = pf
	handler = fn
	app.AsyncSafe(loop)
}

func loop() {
	for {
		if app.CTX.Err() != nil {
			return
		}
		ss, err := rc.List(pfname + "/Query").BLPop(1).Strings()
		if err != nil {
			if err != redis.ErrNil {
				log.Err(err)
				time.Sleep(time.Second)
			}
			continue
		}
		if len(ss) != 2 {
			log.Err("invalid blpop:", ss)
			continue
		}
		if app.CTX.Err() != nil {
			rc.List(pfname + "/Query").LPush(ss[1]).Error()
			return
		}
		m := &msg.Msg{}
		if err := m.Decode(ss[1]); err != nil {
			log.Err(err, ss[1])
			continue
		}
		w.Handle(m)
	}
}

func handleMsg(m *msg.Msg) {
	exist, err := handler(m.Account)
	if err != nil {
		m.Result = -1
		m.Err = err.Error()
	} else {
		if exist {
			m.Result = 1
		} else {
			m.Result = 0
		}
	}
	if err := rc.List(pfname + "/Result").RPush(m.Encode()).Error(); err != nil {
		log.Err(err)
	}
}
